#include <gtest/gtest.h>
#include <gmock/gmock.h>
#include <vendor_global.h>
#include <random>
#include <vector>
#include "downlink_connect.h"
#include <dbms/dbms_channel.h>
#include "serial/serial.h"
#include <thread>
#include <yaml-cpp/yaml.h>
#include <future>

#define GTEST_OUT std::cout
#define GTEST_ENDL std::endl
#define TS(func)                                                               \
	do {                                                                       \
		struct timeval t1, t2;                                                 \
		gettimeofday(&t1, NULL);                                               \
		func;                                                                  \
		gettimeofday(&t2, NULL);                                               \
		printf(#func ": %.3fmsecs\n", (t2.tv_sec - t1.tv_sec) * 1000.0 +       \
										  (t2.tv_usec - t1.tv_usec) / 1000.0); \
	} while (0)


std::vector<std::string> g_params;

namespace downlink {
class DownlinkConnectTest : public ::testing::Test {
public:
	void SetUp() override
	{
		uvLoop_ = uv_loop_new();

		uv_timer_init(uvLoop_, &timer_);
	}
	void TearDown() override { uv_loop_delete(uvLoop_); }

	void SetupTcpServer(int32_t port)
	{
		uv_tcp_init(uvLoop_, &tcpServer_);

		struct sockaddr_in addr;
		uv_ip4_addr("0.0.0.0", port, &addr);
		uv_tcp_bind(&tcpServer_, (const struct sockaddr *)&addr, 0);

		if (uv_listen((uv_stream_t *)&tcpServer_, 128,
				[](uv_stream_t *server, int status) {
					if (status < 0) {
						fprintf(stderr, "Connection error: %s\n",
							uv_strerror(status));
						return;
					}

					// 初始化客户端句柄
					uv_tcp_t *client = (uv_tcp_t *)malloc(sizeof(uv_tcp_t));
					uv_tcp_init(server->loop, client);

					// 接受连接
					if (uv_accept(server, (uv_stream_t *)client) == 0) {
						printf("New client connected\n");
						uv_read_start((uv_stream_t *)client,
							[](uv_handle_t *handle, size_t suggested_size,
								uv_buf_t *buf) {
								buf->base = (char *)malloc(
									suggested_size); // 动态分配内存[2,6](@ref)
								buf->len = suggested_size;
							},
							[](uv_stream_t *client, ssize_t nread,
								const uv_buf_t *buf) {
								if (nread > 0) {
									printf("Received %zd bytes: %.*s\n", nread,
										(int)nread, buf->base);
									// 示例：原样回显数据
									uv_write_t *req = (uv_write_t *)malloc(
										sizeof(uv_write_t));
									uv_buf_t wrbuf =
										uv_buf_init(buf->base, nread);
									uv_write(req, client, &wrbuf, 1,
										[](uv_write_t *req, int status) {
											printf("status: %d\n", status);
											// free(req);
										});
								} else if (nread < 0) {
									if (nread != UV_EOF) {
										fprintf(stderr, "Read error: %s\n",
											uv_strerror(nread));
									}
									uv_close((uv_handle_t *)client,
										[](uv_handle_t *handle) {
											free(handle);
										});
								}
								free(buf->base); // 释放缓冲区内存[2,6](@ref)
							});
					} else {
						uv_close((uv_handle_t *)client, NULL);
					}
				})) {
			fprintf(stderr, "Listen error\n");
			return;
		}
	}


	bool GetChannelFromFile(dbms::DeviceChannel_t &deviceChannel)
	{
		YAML::Node config;
		try {
			config = YAML::LoadFile(VENDOR_TEST_PATH + "/downlink_test.yaml");
		} catch (...) {
			return false;
		}

		YAML::Node testNode = config["test"];
		times_ = testNode["times"].as<int32_t>();
		std::string testCmd = testNode["cmd"].as<std::string>();
		sendBuf_ = utils::strtohexcmd(testCmd);

		YAML::Node channelNode = config["channel"];
		deviceChannel.set_id(channelNode["id"].as<int32_t>());
		deviceChannel.set_timeout(channelNode["waitTime"].as<int32_t>());
		deviceChannel.set_interval(channelNode["intervalTime"].as<int32_t>());
		deviceChannel.set_reportperiod(
			channelNode["reportPeriod"].as<int32_t>());
		std::string type = channelNode["type"].as<std::string>();
		if (type == "Serial") {
			deviceChannel.set_ctype(dbms::CT_COM);
			::dbms::Serial_t *com = deviceChannel.mutable_com();
			YAML::Node params = channelNode["params"];
			com->set_name(params["name"].as<std::string>());
			com->set_baudrate(params["baudRate"].as<int32_t>());
			com->set_databit(params["dataBits"].as<int32_t>());
			com->set_stopbit(params["stopBits"].as<int32_t>());
			com->set_parity(params["parity"].as<int32_t>());
		} else {
			deviceChannel.set_ctype(dbms::CT_TCP);
			YAML::Node params = channelNode["params"];
			::dbms::Socket_t *socket = deviceChannel.mutable_socket();
			socket->set_ip(params["ip"].as<std::string>());
			socket->set_port(params["port"].as<int32_t>());
		}

		return true;
	}

	std::vector<uint8_t> OnDownlink(std::shared_ptr<DownlinkConnect> downlink,
		const std::vector<uint8_t> &sendCmd, int32_t timeout)
	{
		std::vector<uint8_t> recvCmd;
		std::atomic<bool> setFlag{false};
		std::promise<bool> finish;
		auto future = finish.get_future();
		downlink->OnReceive([&](uint8_t recv[], int len) {
			printf("on receive\n");
			recvCmd.reserve(recvCmd.size() + len * sizeof(uint));
			std::copy(reinterpret_cast<const uint8_t *>(recv),
				reinterpret_cast<const uint8_t *>(recv + len),
				std::back_inserter(recvCmd));
			if (!setFlag.exchange(true)) {
				finish.set_value(true);
			}
		});
		downlink->OnError([&](int32_t err) {
			printf("on error, %d\n", err);
			if (!setFlag.exchange(true)) {
				finish.set_value(err == 0);
			}
		});

		bool result = downlink->Send(uvLoop_, sendCmd.data(), sendCmd.size());
		if (result == false) {
			return recvCmd;
		}
		future.wait_for(std::chrono::milliseconds(timeout));
		downlink->Stop();
		return recvCmd;
	}
	uv_loop_t *uvLoop_;
	uv_timer_t timer_;
	uv_tcp_t tcpServer_;
	std::thread threadTask_;
	std::condition_variable condWait_;
	std::mutex taskMutex_;
	int times_ = 1;
	std::vector<uint8_t> sendBuf_;
};


TEST_F(DownlinkConnectTest, Channel)
{
	dbms::DeviceChannel_t deviceChannel;
	if (!GetChannelFromFile(deviceChannel)) {
		return;
	}
	if (deviceChannel.ctype() == dbms::CT_TCP) {
		SetupTcpServer(deviceChannel.socket().port());
	}
	std::shared_ptr<DownlinkConnect> downlink =
		DownlinkConnect::ConnectFactory(deviceChannel);

	std::thread threadTest([&]() {
		std::vector<uint8_t> recv;
		TS(recv = OnDownlink(downlink, sendBuf_, deviceChannel.timeout()));
		EXPECT_GE(recv.size(), 1);
		uv_stop(uvLoop_);
		uv_walk(
			uvLoop_,
			[](uv_handle_t *handle, void *arg) {
				if (!uv_is_closing(handle)) {
					uv_close(handle, nullptr);
				}
			},
			nullptr);
	});
	threadTest.detach();
	uv_timer_start(&timer_, [](uv_timer_t *handle) {}, 5000, 5000);
	uv_run(uvLoop_, UV_RUN_DEFAULT);

	printf("loop exit\n");
}

TEST_F(DownlinkConnectTest, Channel_PressTest)
{
	dbms::DeviceChannel_t deviceChannel;
	if (!GetChannelFromFile(deviceChannel)) {
		return;
	}
	if (deviceChannel.ctype() == dbms::CT_TCP) {
		SetupTcpServer(deviceChannel.socket().port());
	}
	std::shared_ptr<DownlinkConnect> downlink =
		DownlinkConnect::ConnectFactory(deviceChannel);


	std::thread threadTest([&]() {
		for (int i = 0; i < times_; ++i) {
			std::vector<uint8_t> recv;
			TS(recv = OnDownlink(downlink, sendBuf_, deviceChannel.timeout()));
			EXPECT_GE(recv.size(), 1);
			std::this_thread::sleep_for(
				std::chrono::milliseconds(deviceChannel.interval()));
		}
		uv_stop(uvLoop_);
		uv_walk(
			uvLoop_,
			[](uv_handle_t *handle, void *arg) {
				if (!uv_is_closing(handle)) {
					uv_close(handle, nullptr);
				}
			},
			nullptr);
	});
	threadTest.detach();
	uv_timer_start(&timer_, [](uv_timer_t *handle) {}, 5000, 5000);
	uv_run(uvLoop_, UV_RUN_DEFAULT);
}
} // namespace downlink


int main(int argc, char *argv[])
{
	utils::set_enable(true);
	for (int i = 0; i < argc; ++i) {
		g_params.emplace_back(argv[i]);
	}
	::testing::InitGoogleTest(&argc, argv);
	return RUN_ALL_TESTS();
}